package qq2564874169.tuaotuao.fx.lock.zookeeper;

import org.apache.zookeeper.*;
import qq2564874169.tuaotuao.fx.*;
import qq2564874169.tuaotuao.fx.concurrent.FxCountDownLatch;
import qq2564874169.tuaotuao.fx.lock.AbortTimeoutException;
import qq2564874169.tuaotuao.fx.lock.Lockspace;
import qq2564874169.tuaotuao.fx.lock.LockspaceException;
import qq2564874169.tuaotuao.fx.utils.ExUtil;

import java.util.concurrent.TimeUnit;

public class ZkLockspace implements Lockspace {
    private static String rootNode;
    private ZooKeeper zk;

    public static void createRootNode(ZooKeeper zk) {
        createRootNode(zk, "/FxLockspace");
    }

    public static void createRootNode(ZooKeeper zk, String nodeName) {
        try {
            if (zk.exists(nodeName, false) == null) {
                zk.create(nodeName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            rootNode = nodeName;
        } catch (KeeperException | InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    public ZkLockspace(ZooKeeper zk) {
        this.zk = zk;
    }

    private static String fullPath(String path) {
        FxAssert.isNotNull(path);
        FxAssert.isNotNull(rootNode, "请先调用createRootNode创建根节点");
        if (path.startsWith("/")) {
            return rootNode + path;
        } else {
            return rootNode + "/" + path;
        }
    }

    @Override
    public FxAutoCloseable queueLock(String key, Integer abortTimeout) throws AbortTimeoutException {
        var st = new FxDate();
        while (FxEnv.isShutdown() == false) {
            try {
                var k = zk.create(fullPath(key), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                return () -> {
                    try {
                        zk.delete(k, -1);
                    } catch (InterruptedException | KeeperException e) {
                        //...
                    }
                };
            } catch (KeeperException e) {
                if (e.code().equals(KeeperException.Code.NODEEXISTS)) {
                    if (abortTimeout != null) {
                        if (st.addSecond(abortTimeout).yyyyMMddHHmmss() <= new FxDate().yyyyMMddHHmmss()) {
                            throw new AbortTimeoutException(key);
                        } else {
                            var t = st.addSecond(abortTimeout).getTime();
                            t = (t - new FxDate().getTime()) / 3;
                            Task.delay(Math.max(100, t));
                        }
                    } else {
                        Task.delay(100, 1500);
                    }
                } else {
                    throw ExUtil.toRuntime(e);
                }
            } catch (InterruptedException e) {
                throw ExUtil.toRuntime(e);
            }
        }
        throw new RuntimeException("进程已退出");
    }

    @Override
    public FxAutoCloseable exLock(String key) throws LockspaceException {
        try {
            var k = zk.create(fullPath(key), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            return () -> {
                try {
                    zk.delete(k, -1);
                } catch (InterruptedException | KeeperException e) {
                    //...
                }
            };
        } catch (KeeperException e) {
            if (e.code().equals(KeeperException.Code.NODEEXISTS)) {
                throw new LockspaceException(key);
            } else {
                throw ExUtil.toRuntime(e);
            }
        } catch (InterruptedException e) {
            throw ExUtil.toRuntime(e);
        }
    }

    @Override
    public FxAutoCloseable concurrentLock(String key, int max, Integer abortTimeout) throws AbortTimeoutException {
        var pkey = fullPath(key);
        var nkey = "lock";
        var ckey = pkey + "/" + nkey;
        var st = new FxDate();
        while (FxEnv.isShutdown() == false) {
            try {
                var currkey = zk.create(ckey, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                var num = Integer.parseInt(currkey.substring(ckey.length() + 1));
                while (FxEnv.isShutdown() == false) {
                    var childs = FxStream.create(zk.getChildren(pkey, false)).toMap(i -> Integer.parseInt(i.substring(nkey.length() + 1)));
                    var pre = FxStream.create(childs.keySet()).filter(i -> i < num);
                    if (pre.count() >= max) {
                        var wkey = pre.sort().reverse().skip(max - 1).get(0);
                        var latch = new FxCountDownLatch(1);
                        var stat = zk.exists(fullPath(childs.get(wkey)), e -> {
                            if (e.getType() == Watcher.Event.EventType.NodeDeleted) {
                                latch.countDown();
                            }
                        });
                        if (stat != null) {
                            if (abortTimeout != null) {
                                try {
                                    latch.await(abortTimeout.longValue(), TimeUnit.SECONDS);
                                } catch (Exception e) {
                                    if (st.addSecond(abortTimeout).yyyyMMddHHmmss() < new FxDate().yyyyMMddHHmmss()) {
                                        throw new AbortTimeoutException(key);
                                    }
                                    throw ExUtil.toRuntime(e);
                                }
                            } else {
                                latch.await();
                            }
                            break;
                        } else {
                            continue;
                        }
                    }
                    break;
                }
                return () -> {
                    try {
                        zk.delete(currkey, -1);
                    } catch (InterruptedException | KeeperException e) {
                        //...
                    }
                };
            } catch (KeeperException e) {
                if (e.code().equals(KeeperException.Code.NONODE)) {
                    try {
                        zk.create(pkey, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);
                    } catch (KeeperException ex) {
                        if (ex.code().equals(KeeperException.Code.NODEEXISTS) == false) {
                            throw ExUtil.toRuntime(e);
                        }
                    } catch (InterruptedException ex) {
                        throw ExUtil.toRuntime(e);
                    }
                } else {
                    throw ExUtil.toRuntime(e);
                }
            } catch (InterruptedException e) {
                throw ExUtil.toRuntime(e);
            }
        }
        throw new RuntimeException("进程已退出");
    }
}
